//
// Created by 33145 on 2023/6/8.
//

#include "rpc_dispatcher.h"
#include "rpc_controller.h"
#include "../common/rocket.h"

#include <google/protobuf/service.h>
#include <google/protobuf/descriptor.h>
#include <google/protobuf/message.h>
#include "../tcp/tcp_connection.h"

namespace rocket {

  static RPCDispatcher::ptr g_rpc_dispatcher = nullptr;

#define DELETE_RESOURCE(XX) \
  if (XX != NULL) { \
    delete XX;      \
    XX = NULL;      \
  }                 \

  void RPCDispatcher::dispatch(const AbstractProtocol::ptr& request, const AbstractProtocol::ptr& response, TCPConnection* connection) {
    auto req_protocol = std::dynamic_pointer_cast<TinyPBProtocol>(request);
    auto rsp_protocol = std::dynamic_pointer_cast<TinyPBProtocol>(response);
    // 方法名： OrderService.method_name 需要解析
    std::string method_full_name = req_protocol->_method_name;
    std::string service_name;
    std::string method_name;
    if (!parseServiceFullName(method_full_name, service_name, method_name)) {
      setTinyPBError(rsp_protocol, ERROR_PARSE_SERVICE_NAME, "parse service name error");
      return;
    }
    auto it = _service_map.find(service_name);
    if (it == _service_map.end()) {
      LOG_ERROR("%s | service name[%s] not found", req_protocol->_msg_id.c_str(), service_name.c_str());
      setTinyPBError(rsp_protocol, ERROR_SERVICE_NOT_FOUND, "service not found");
      return;
    }
    service_ptr service = it->second;
    auto method = service->GetDescriptor()->FindMethodByName(method_name);
    if (method == nullptr) {
      LOG_ERROR("%s | method name[%s] not found in service[%s]", req_protocol->_msg_id.c_str(), method_name.c_str(), service_name.c_str());
      setTinyPBError(rsp_protocol, ERROR_SERVICE_NOT_FOUND, "method not found");
      return;
    }
    auto req_msg = service->GetRequestPrototype(method).New();
    // 反序列化，将 pb_data 反序列化为 req_msg
    if (!req_msg->ParseFromString(req_protocol->_pb_data)) {
      LOG_ERROR("%s | deSerialize error", req_protocol->_msg_id.c_str(), method_name.c_str(), service_name.c_str());
      setTinyPBError(rsp_protocol, ERROR_FAILED_DESERIALIZE, "deSerialize error");
      DELETE_RESOURCE(req_msg);
      return;
    }

    // 获得response
    auto rsp_msg = service->GetResponsePrototype(method).New();

    RPCController::ptr RPCctl = std::make_shared<RPCController>();
    // 拿到两端地址 msgID
    RPCctl->SetLocalAddr(connection->getLocalAddr());
    RPCctl->SetPeerAddr(connection->getPeerAddr());
    RPCctl->SetMsgId(req_protocol->_msg_id);


    service->CallMethod(method, RPCctl.get(), req_msg, rsp_msg, nullptr);  // 调用RPC方法 （在.proto中定义）



    if (!rsp_msg->SerializeToString(&(rsp_protocol->_pb_data))) {
      LOG_ERROR("%s | Serialize error, origin message [%s]", req_protocol->_msg_id.c_str(),
               rsp_msg->ShortDebugString().c_str());
      setTinyPBError(rsp_protocol, ERROR_FAILED_SERIALIZE, "Serialize error");
      DELETE_RESOURCE(req_msg);
      DELETE_RESOURCE(rsp_msg);
      return;
    }
    rsp_protocol->_err_code = 0;
    rsp_protocol->_msg_id = req_protocol->_msg_id;
    LOG_INFO("%s | get rpc response [%s]", req_protocol->_msg_id.c_str(), rsp_msg->ShortDebugString().c_str());
    DELETE_RESOURCE(req_msg);
    DELETE_RESOURCE(rsp_msg);
  }

  void RPCDispatcher::registerService(const RPCDispatcher::service_ptr& service) {
    _service_map[service->GetDescriptor()->full_name()] = service;
  }

  bool RPCDispatcher::parseServiceFullName(std::string &full_name, std::string &service_name, std::string &method_name) {
    if (full_name.empty()) {
      LOG_ERROR("full name empty");
      return false;
    }
    size_t i = full_name.find_first_of('.');
    if (i == std::string::npos) {
      LOG_ERROR("not find . in full name [%s]", full_name.c_str());
      return false;
    }
    service_name = full_name.substr(0, i);
    method_name = full_name.substr(i + 1, full_name.length() - i - 1);

    LOG_INFO("parse service_name[%s] and method_name[%s] from full name [%s]", service_name.c_str(), method_name.c_str(),full_name.c_str());

    return true;
  }

  void RPCDispatcher::setTinyPBError(const TinyPBProtocol::ptr & msg, int32_t err_code, const std::string& err_info) {
    msg->_err_code = err_code;
    msg->_err_info = err_info;
    msg->_err_info_len = err_info.length();
  }

  RPCDispatcher::ptr RPCDispatcher::GetRPCDispatcher() {
    if (g_rpc_dispatcher == nullptr) {
      g_rpc_dispatcher = std::make_shared<RPCDispatcher>();
    }
    return g_rpc_dispatcher;
  }
}